package com.mjf.day5

import com.mjf.day3.{SensorReading, SensorSource}
import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

/**
 * 侧输出流
 */
object SideOutputExample {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val inputStream: DataStream[SensorReading] = env.addSource(new SensorSource)

    // 用ProcessFunction的侧输出流实现分流操作
    val highTempStream: DataStream[SensorReading] = inputStream.process(new SplitTempProcessor(30))

    val lowTempStream: DataStream[SensorReading] = highTempStream.getSideOutput(new OutputTag[SensorReading]("low-temp")) // 泛型可以是不同类型
//    val lowTempStream: DataStream[SensorReading] = highTempStream.getSideOutput(new OutputTag[(String, Long, Double)]("low-temp")) // 泛型可以是不同类型

    highTempStream.print("high")
    lowTempStream.print("low")

    env.execute("SideOutputExample")

  }
}

// 自定义ProcessFunction,用于区分高低温数据
class SplitTempProcessor(threshold: Double) extends ProcessFunction[SensorReading, SensorReading] {
  override def processElement(value: SensorReading, ctx: ProcessFunction[SensorReading, SensorReading]#Context, out: Collector[SensorReading]): Unit = {
     // 判断当前数据的温度值，如果大于阈值，输出到主流。如果小于阈值，输出到侧输出流
    if(value.temperature > threshold) {
      out.collect(value)
    } else {
      ctx.output(new OutputTag[SensorReading]("low-temp"), value)
//      ctx.output(new OutputTag[(String, Long, Double)]("low-temp"), (value.id, value.timestamp, value.temperature))
    }
  }
}
